#pragma once 

/*这文件是对 RabbitMQ 的二次封装。*/

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include "logger.hpp"

namespace ChatSystem
{

class MQClient
{
public:
    using MessageCallback = std::function<void(const char* body, size_t)>;
    using ptr = std::shared_ptr<MQClient>;
    MQClient(const std::string& user, 
        const std::string& password, 
        const std::string& host)
        {
            _loop = EV_DEFAULT;
            _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
            //amqp://root:123456@127.0.0.1:5672/
            std::string url = "amqp://" + user + ":" + password + "@" + host + "/";
            AMQP::Address address(url);
            _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
            _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

            _loop_thread = std::thread([this](){
                ev_run(_loop, 0);   //开始运行I/O事件监控---创建一个新的线程进行运行
            });
        }
    ~MQClient()
    {
        //进行I/O事件监控的不是当前的执行流，所以要进行异步事件的通知
        ev_async_init(&_async_watcher, watcher_callback);
        ev_async_start(_loop, &_async_watcher);
        ev_async_send(_loop, &_async_watcher);
        _loop_thread.join();
        _loop = nullptr;    //在RabbitMQ的代码中已经做了释放的动作，这里不需要重复释放，把_loop置为空就可以了
    }
    //负责声明交换机、声明队列、交换机和队列进行绑定
    void declareComponents(const std::string& exchange, 
        const std::string& queue,
    const std::string& routing_key = "routing_key",
    AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct)
    {
        //4. 声明交换机
        _channel->declareExchange(exchange, exchange_type)  
        .onError([](const char* message){
            LOG_ERROR("声明交换机失败：{}", message);
            exit(0);
        })
        .onSuccess([exchange](){
            LOG_INFO("{} 声明交换机成功!", exchange);
        });
        //5. 声明队列
        _channel->declareQueue(queue)
        .onError([](const char* message){
            LOG_ERROR("声明队列失败：{}", message);
            exit(0);
        })
        .onSuccess([queue](){
            LOG_INFO("{} 声明队列成功!", queue);
        });
        //6. 针对交换机和队列进行绑定
        _channel->bindQueue(exchange, queue, routing_key)
        .onError([exchange, queue](const char* message){
            LOG_ERROR("{} - {} 绑定失败：{}", exchange, queue, message);
            exit(0);
        })
        .onSuccess([exchange, queue](){
            LOG_INFO("{} - {} 绑定成功!", exchange, queue);
        });
    }

    //消息的发布
    bool publish(const std::string& exchange,
        const std::string& msg,
        const std::string& routing_key = "routing_key")
        {
            LOG_DEBUG("向交换机 {} 发布消息！", exchange);
            bool ret = _channel->publish(exchange, routing_key, msg);
            if(ret == false)
            {
                LOG_ERROR("向交换机 {} 发布消息失败！", exchange);
                return false;
            }
            return true;
        }

    //消息的订阅
    void consume(const std::string& queue, 
        const MessageCallback& cb)
        {
            LOG_DEBUG("从消息队列 {} 中开始订阅消息！", queue);
            _channel->consume(queue, queue)        //返回值是一个DeferredConsumer 类型
            .onReceived([this, cb](const AMQP::Message& message, uint64_t deliverTag, bool redelivered){
                cb(message.body(), message.bodySize());
                //消息确认
                _channel->ack(deliverTag);
            })
            .onError([queue](const char* message){
                LOG_ERROR("订阅 {} 队列消息失败！", queue);
                exit(0);
            });     //返回中是一个AMQP::Deferred;  DeferredConsumer是 AMQP::Deferred派生的子类
        }
private:
    static void watcher_callback(struct ev_loop* loop, ev_async* watcher, int32_t events)
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_async _async_watcher;
    struct ev_loop* _loop;  //I/O事件监控句柄
    std::unique_ptr<AMQP::LibEvHandler> _handler;    //libEvHandler句柄
    std::unique_ptr<AMQP::TcpConnection> _connection;    //连接
    std::unique_ptr<AMQP::TcpChannel> _channel;  //信道
    std::thread _loop_thread;   //当前的线程
};

}